package com.zjl.Kafka.第01章_介绍和安装使用;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 查询  此值
 * echo $JAVA_HOME
 *
 * 刷新 profile文件（环境变量）
 * source /etc/profile
 *
 * 1.要配置Kafka以允许外网访问，您需要修改Kafka的配置文件server.properties。
 * 2.找到listeners配置项，默认可能是listeners=PLAINTEXT://localhost:9092，
 *      将其修改为listeners=PLAINTEXT://0.0.0.0:9092。
 *      这样Kafka就会在所有网络接口上监听9092端口。
 * 3.确保advertised.listeners配置项被设置为Kafka将向客户端宣告的可访问的地址和端口。
 *      例如，如果Kafka部署在一个云服务上，并且该服务有一个公共IP，
 *      则应设置为advertised.listeners=PLAINTEXT://<公共IP>:9092。
 * 4.修改Kafka的防火墙规则，以允许外部访问9092端口。
 * 5.如果使用了Kafka集群，确保所有的broker节点的配置都进行了相应的修改。
 * 6.重启Kafka服务以使配置生效。
 *
 * 而且  版本也要对应，Zookeeper 的最好使用 Kafka 自带的 Zookeeper 否则，会出现预料之外的问题
 *
 */
public class E_Java连接kafka {
    @Test
    public  void 生产者() {
        Map<String,Object> config = new HashMap<>();
//        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9092");//连接地址
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9091,82.157.71.243:9092,82.157.71.243:9093");//连接地址
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//key 的 序列化
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//value 的 序列化
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        //创建生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(config);
        for (int i = 0; i < 10; i++) {

            //创建数据
            ProducerRecord<String,String> record = new ProducerRecord<>(
                    "zjl1"//主题名称
                    ,"key"//数据的key
                    ,"value" + i//数据的value
            );
            //将数据发送到kafka
            kafkaProducer.send(record);
            System.out.println(record);
            try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { throw new RuntimeException(e); }
        }
        try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { throw new RuntimeException(e); }
        //关闭对象
        kafkaProducer.close();
    }

    @Test
    public  void 消费者() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9092");//连接地址
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9091,82.157.71.243:9092,82.157.71.243:9093");//连接地址
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//key 的 反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//value 的 反序列化
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "java001");//消费者组的名称
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);//自动提交偏移量
        config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);//自动提交时间
        //创建消费者对象
        KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<>(config);
        //订阅主题
        List<String> topics  = new ArrayList<>();
        topics.add("java001");
        topics.add("zjl1");
        topics.add("zjl001");
        kafkaConsumer.subscribe(topics);

        List<TopicPartition> list = new ArrayList<>();
        list.add(new TopicPartition("zjl1",0));
        list.add(new TopicPartition("java001",0));
//        kafkaConsumer.assign(list);

        while (true){

            //从生产者拉取数据
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofSeconds(3));
            System.out.println(polls.partitions().size());
            for (TopicPartition tops:polls.partitions()){

                List<ConsumerRecord<String, String>> records = polls.records(tops);
                for (ConsumerRecord<String, String> pool:records){

                    //消费的数据案例
                    //ConsumerRecord(topic = test2, partition = 0, leaderEpoch = 0, offset = 29, CreateTime = 1725273770592, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = key9, value = value9)
                    System.out.println(pool);
                }
            }
        }

        //关闭消费者
//        kafkaConsumer.close();

    }
}
